package com.streaming.demo

import org.apache.spark.SparkConf
import kafka.serializer.StringDecoder
import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

/**
  * @author td
  * @date 2018/4/3
  */
object SparkStreamingKafkaDemo {


  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("kafkaDemo");

    val ssc = new StreamingContext(conf,Seconds(8));
    ssc.checkpoint("checkpoint")
  //  KafkaStreamingData(ssc);
    kafkaStreamingDataDirect(ssc);
  }

  def kafkaStreamingDataDirect(context: StreamingContext): Unit = {

    val topicsSet = Set("sparkTopic");
    val kafkaParams = Map[String,String]("metadata.broker.list"->"172.16.88.170:9092");
    val messages = KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](context,kafkaParams,topicsSet);
    val lines = messages.map(value=>value._2);
    val words = lines.flatMap(_.split(" ")).map((_,1));
    val wordCounts = words.reduceByKeyAndWindow(_+_,Seconds(24));
    wordCounts.print();
    context.start();
    context.awaitTermination();
  }

  def KafkaStreamingData(streamingContext: StreamingContext): Unit ={

    val kafkaConfig =  Array[String]("172.16.88.170:2181","sparkgroup","sparkTopic");


    val topicSet =  Set(kafkaConfig(2))

    val topicMap = Map(kafkaConfig(2)->2)

    val kafkaParams = Map[TopicPartition,String](new TopicPartition("sparkTopic",2)->"111");

    val kafkaStream = KafkaUtils.createStream(streamingContext,kafkaConfig(0),kafkaConfig(1),topicMap);

    val lines = kafkaStream.map(value=>value._2);

    val wordCount = lines.flatMap(_.split(" ")).map((_,1));

   // val reduceBykey = wordCount.reduceByKey(_+_);

    val updateFunc = (values:Seq[Int],state:Option[Int]) => {

      val currentCount = values.sum;
      val privioutCount = state.getOrElse(0);

      Some(currentCount+privioutCount);
    }
   // val result = reduceBykey.updateStateByKey(updateFunc);
    val result = wordCount.reduceByKeyAndWindow(_+_,Seconds(24))

    result.print();

    streamingContext.start();

    streamingContext.awaitTermination();
  }


}
